package mapred;

import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.Semaphore;

import mapred.config.AppConfig;
import mapred.config.CombineConfig;
import mapred.config.InputConfig;
import mapred.config.JobConfig;
import mapred.config.MapConfig;
import mapred.config.OutputConfig;
import mapred.config.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;

public abstract class Job extends Thread {

    private Semaphore sem;

    private String jobName;

    private AppConfig appConfig;

    public Job(String jobName, String... settings) {
        this(jobName, new Semaphore(1), settings);
    }

    public Job(String jobName, Semaphore sem, String... settings) {
        super();

        this.jobName = jobName.replaceAll("\\s", "_");
        this.sem = sem;

        appConfig = new AppConfig();
        appConfig.set("-timeout", 600000);
        appConfig.set("-vmargs", "-Xmx2g");
        appConfig.set("-overwrite", "no");
        for (String setting : settings) {
            int pos = setting.indexOf("=");
            if (pos >= 0) {
                appConfig.set(setting.substring(0, pos), setting
                        .substring(pos + 1));
            } else {
                appConfig.set(null, setting);
            }
        }
    }

    public void run() {
        try {
            sem.acquire();

            JobConfig config = new JobConfig();
            config.getJobConf().setJobName(jobName);
            config.set("mapred.task.timeout", appConfig.get("-timeout"));
            config.set("mapred.child.java.opts", appConfig.get("-vmargs"));
            FileSystem fs = config.getFileSystem();
            setUp(config, fs);

            setOutput(config.getJobConf(), fs);
            setReduce(config.getJobConf());
            setCombine(config.getJobConf());
            setMap(config.getJobConf());
            setInput(config.getJobConf(), fs);

            System.out.println(jobName);
            System.out.println(appConfig);
            RunningJob job = JobClient.runJob(config.getJobConf());
            tearDown(config, fs, job);

            sem.release();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void setOutput(JobConf conf, FileSystem fs) throws IOException {
        OutputConfig oc = this.getOutputConfig(fs);

        conf.setOutputFormat(oc.getOutputFormat());
        conf.setOutputKeyClass(oc.getOutputKeyClass());
        conf.setOutputValueClass(oc.getOutputValueClass());

        Path outputPath = new Path(oc.getOutputPath());
        if (appConfig.containsKey("-output")) {
            outputPath = new Path(appConfig.get("-output"));
        }
        FileOutputFormat.setOutputPath(conf, outputPath);
        System.out.println("Output Path: " + outputPath);
        if (fs.exists(outputPath)) {
            if (appConfig.get("-overwrite").toLowerCase().startsWith("y")) {
                fs.delete(outputPath, true);
            } else {
                System.out.println("Output Path " + outputPath.toString()
                        + " already exists, overwrite it?(y/N)");
                Scanner sc = new Scanner(System.in);
                if (sc.next().trim().toLowerCase().startsWith("y")) {
                    fs.delete(outputPath, true);
                } else {
                    System.exit(0);
                }
            }
        }

        if (appConfig.containsKey("-compress")) {
            String outputCompress = appConfig.get("-compress");
            if (!"null".equalsIgnoreCase(outputCompress)) {
                conf.setCompressMapOutput(true);
                if ("default".equalsIgnoreCase(outputCompress)) {
                    FileOutputFormat.setOutputCompressorClass(conf,
                            DefaultCodec.class);
                } else if ("gzip".equalsIgnoreCase(outputCompress)) {
                    FileOutputFormat.setOutputCompressorClass(conf,
                            GzipCodec.class);
                }
            }
        } else {
            if (oc.isCompressOutput()) {
                FileOutputFormat.setCompressOutput(conf, true);
                FileOutputFormat.setOutputCompressorClass(conf, oc
                        .getOutputCompressorClass());
            }
        }
    }

    private void setReduce(JobConf conf) {
        ReduceConfig rc = this.getReduceConfig();
        conf.setReducerClass(rc.getReducerClass());
        if (appConfig.containsKey("-reduce")) {
            conf.setNumReduceTasks(Integer.parseInt(appConfig.get("-reduce")));
        } else {
            conf.setNumReduceTasks(rc.getNumReduceTasks());
        }
        conf.setMaxReduceAttempts(10);
    }

    private void setCombine(JobConf conf) {
        CombineConfig cc = this.getCombineConfig();
        if (cc.getCombinerClass() != null) {
            conf.setCombinerClass(cc.getCombinerClass());
        }
    }

    private void setMap(JobConf conf) {
        MapConfig mc = this.getMapConfig();
        conf.setMapperClass(mc.getMapperClass());
        if (appConfig.containsKey("-map")) {
            conf.setNumMapTasks(Integer.parseInt(appConfig.get("-map")));
        } else {
            conf.setNumMapTasks(mc.getNumMapTasks());
        }
        conf.setMapOutputKeyClass(mc.getOutputKeyClass());
        conf.setMapOutputValueClass(mc.getOutputValueClass());
        conf.setMapOutputCompressorClass(DefaultCodec.class);

        conf.setMaxMapTaskFailuresPercent(1);
        conf.setMaxMapAttempts(10);
    }

    private void setInput(JobConf conf, FileSystem fs) throws IOException {
        InputConfig ic = this.getInputConfig(fs);
        conf.setInputFormat(ic.getInputFormat());
        String[] paths = ic.getInputPaths();
        if (appConfig.containsKey("-input")) {
            paths = appConfig.get("-input").split(",");
        }
        for (String path : paths) {
            Path inputpath = new Path(path);
            if (fs.exists(inputpath)) {
                FileInputFormat.addInputPath(conf, inputpath);
                System.out.println("Input Path: " + inputpath);
            } else {
                System.out.println("Cannot find path " + inputpath);
            }
        }
    }

    public void submit() throws Exception {
        this.start();
        Thread.sleep(1000);
        sem.acquire();
    }

    protected void setAppConfig(String key, Object value) {
        this.appConfig.set(key, value);
    }
    
    protected String getAppConfig(String key) {
        return this.appConfig.get(key);
    }
    
    public static void buildXML(Class<?> cls) throws Exception {
        AntBuilder.buildXML(cls);
    }
    
    protected abstract void tearDown(JobConfig config, FileSystem fs, RunningJob job)
            throws IOException;

    protected abstract void setUp(JobConfig config, FileSystem fs)
            throws IOException;

    protected abstract OutputConfig getOutputConfig(FileSystem fs)
            throws IOException;

    protected abstract ReduceConfig getReduceConfig();

    protected abstract CombineConfig getCombineConfig();

    protected abstract MapConfig getMapConfig();

    protected abstract InputConfig getInputConfig(FileSystem fs)
            throws IOException;

}
